package io.lvdaxian.rocketmq.learn.demo.mq.pull.http;

import io.lvdaxian.rocketmq.learn.demo.mq.pull.PullExtRocketMQTemplate;
import io.lvdaxian.rocketmq.learn.demo.mq.pull.po.User;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * pull 消费的 测试controller
 *
 * @author lihh
 */
@RestController
@RequestMapping("/pull")
public class PullController {
  
  @Resource
  private PullExtRocketMQTemplate pullExtRocketMQTemplate;
  @Value("${demo.rocketmq.user.topic}")
  private String stringRequestTopic;
  
  private final static AtomicInteger counter = new AtomicInteger(0);
  
  /**
   * 发送消息 方法
   *
   * @param count 自定义长度
   * @author lihh
   */
  @GetMapping("/sendMessage/{count}")
  public void sendMessage(@PathVariable("count") Integer count) throws InterruptedException {
     while (true) {
       for (int i = 0; i < count; i++) {
         
         String idx = String.valueOf(counter.getAndAdd(1));
          // 构建用户
         User user = User.builder()
             .id(idx)
             .name("name" + idx)
             .build();
         
         pullExtRocketMQTemplate.asyncSend(stringRequestTopic, user, new SendCallback() {
           @Override
           public void onSuccess(SendResult sendResult) {
           
           }
           
           @Override
           public void onException(Throwable throwable) {
           
           }
         });
       }
       
       Thread.sleep(1000);
     }
  }
}
